{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 实验四任务阶段"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## duty1 \n",
    "question1  思路：在对文件读入之后，将数据（经过列筛选）转换为pandas.dataframe，通过设置初始时间和时间数组，对不同年份的交易量进行统计，最后使用groupby分组和sort_values进行排序。\n",
    "\n",
    "question2  思路：将数据（经过列筛选）转换为pandas.dataframe，使用itertuples（）制造迭代器（效率高于iterrows（）），计算每行数据的差值，记为difference，而后对整体数据进行排序，挑选前十行进行输出\n",
    "\n",
    "**输出数据的第一列是数据在源数据中的index ，为了方便进行数据比对，检验程序的正确性，所以没有进行删除处理**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "------------the result of question1,duty 1--------------------\n",
      "      year stock_symbol stock_volume\n",
      "2823  2000         ISRL  13238366000\n",
      "582   2000         ALAN   7920327400\n",
      "2207  2000         ACME   4298582600\n",
      "4715  2000          GAJ   4175617100\n",
      "4296  2000          GNA   2481568700\n",
      "...    ...          ...          ...\n",
      "4025  2009          GRE       101700\n",
      "4839  2009          GJL        86500\n",
      "1353  2009        ASRVP        78800\n",
      "746   2009         AMEN         2400\n",
      "4225  2009          GTC            0\n",
      "\n",
      "[5771 rows x 3 columns]\n",
      "------------the result of question2,duty 1--------------------\n",
      "       exchange stock_symbol       date  stock_price_close  stock_price_open  \\\n",
      "637032   NASDAQ         INFY 2000-02-11             670.06            534.50   \n",
      "637033   NASDAQ         INFY 2000-02-10             528.50            441.50   \n",
      "618948   NASDAQ         INCY 2000-02-17             276.13            224.25   \n",
      "637034   NASDAQ         INFY 2000-02-09             446.00            399.06   \n",
      "930873     NYSE          GTC 2001-01-03             306.23            261.40   \n",
      "931049     NYSE          GTC 2000-04-17             448.37            407.98   \n",
      "394073   NASDAQ         AMCC 2000-04-17             104.62             65.75   \n",
      "41707    NASDAQ         AUDC 2000-03-06             135.13             97.12   \n",
      "636996   NASDAQ         INFY 2000-04-05             197.00            160.00   \n",
      "621677   NASDAQ         ISRL 2009-06-25             103.00             66.06   \n",
      "\n",
      "        difference  \n",
      "637032      135.56  \n",
      "637033       87.00  \n",
      "618948       51.88  \n",
      "637034       46.94  \n",
      "930873       44.83  \n",
      "931049       40.39  \n",
      "394073       38.87  \n",
      "41707        38.01  \n",
      "636996       37.00  \n",
      "621677       36.94  \n"
     ]
    }
   ],
   "source": [
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "import json\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "sc = SparkContext(\"local\", \"duty 1\")\n",
    "columns_json_str = '{\"date\":\"date\",\"stock_symbol\":\"stock_id\",\"stock_volume\":\"volume\"}'\n",
    "columns_dict = json.loads(columns_json_str)\n",
    "spark = SparkSession.builder.getOrCreate()\n",
    "spark.conf.set(\"spark.sql.execution.arrow.pyspark.enabled\", \"true\")\n",
    "df = spark.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('file:///home/allen/myspark/stock_small.csv')\n",
    "df1 = pd.DataFrame(df.toPandas(),columns=columns_dict.keys())\n",
    "firstyear = 2009\n",
    "volume = 0\n",
    "totol_volume_year = pd.DataFrame(columns=['year','stock_symbol','stock_volume'])\n",
    "rownum = 0\n",
    "for row in df1.itertuples():\n",
    "    name = row[2]\n",
    "    if row[1].year == firstyear and row[2] == name:        \n",
    "        volume += int(row[3])\n",
    "    else:\n",
    "        totol_volume_year.loc[rownum,'year'] = firstyear\n",
    "        totol_volume_year.loc[rownum,'stock_symbol'] = name\n",
    "        totol_volume_year.loc[rownum,'stock_volume'] = volume   \n",
    "        rownum+=1     \n",
    "        firstyear = row[1].year\n",
    "        volume = int(row[3])\n",
    "totol_volume_year.loc[rownum,'year'] = firstyear\n",
    "totol_volume_year.loc[rownum,'stock_symbol'] = name\n",
    "totol_volume_year.loc[rownum,'stock_volume'] = volume      \n",
    "groups = totol_volume_year.groupby('year')\n",
    "n = np.arange(2000,2010,1)\n",
    "for i in n:\n",
    "    if i == 2000:\n",
    "        duty11_re = groups.get_group(i).sort_values(by=\"stock_volume\",ascending=False)\n",
    "    else:\n",
    "        duty11_re = duty11_re.append(groups.get_group(i).sort_values(by=\"stock_volume\",ascending=False))\n",
    "columns_json_str = '{\"exchange\":\"exchange\",\"stock_symbol\":\"stock_symbol\",\"date\":\"date\",\"stock_price_close\":\"stock_price_close\",\"stock_price_open\":\"stock_price_open\"}'\n",
    "columns_dict = json.loads(columns_json_str)\n",
    "df2 = pd.DataFrame(df.toPandas(),columns=columns_dict.keys())\n",
    "for row in df2.itertuples():\n",
    "    df2.loc[row[0],'difference'] = float(row[4]) - float(row[5])\n",
    "duty12_re =df2.sort_values(by=['difference'],ascending=False)\n",
    "print('------------the result of question1,duty 1--------------------')\n",
    "print(duty11_re)\n",
    "\n",
    "print('------------the result of question2,duty 1--------------------')\n",
    "print(duty12_re[0:10])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## duty2\n",
    "\n",
    "思路：1，读取文件信息后，使用map函数进行数据切割，使用yspark.sql.row构建数据表（指定列名），在其后使用pyspark.sql.sparksession向数据表中填充数据。  \n",
    "对于question1，采用的sql语句为   \n",
    "    \"SELECT date,stock_symbol,close \\\n",
    "    FROM stocksm \\\n",
    "    WHERE stock_symbol = 'IBM' AND \\\n",
    "    date IN (SELECT date FROM div WHERE symbol = 'IBM')\"   \n",
    "    \n",
    "对于question2， 采用的sql语句为  \n",
    "    \"SELECT year(date) year,avg(close_adj) stock_price_adj_close \\\n",
    "    FROM stocksm \\\n",
    "    WHERE stock_symbol = 'AAPL' \\\n",
    "    GROUP BY year\\\n",
    "    HAVING stock_price_adj_close>50 \\\n",
    "    ORDER BY year \"\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "------------the result of question1,duty 2--------------------\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------+------+\n",
      "|      date|stock_symbol| close|\n",
      "+----------+------------+------+\n",
      "|2000-02-08|         IBM|118.81|\n",
      "|2000-05-08|         IBM|109.75|\n",
      "|2000-08-08|         IBM|118.87|\n",
      "|2000-11-08|         IBM| 100.0|\n",
      "|2001-02-07|         IBM|116.91|\n",
      "|2001-05-08|         IBM| 117.7|\n",
      "|2001-08-08|         IBM|104.19|\n",
      "|2001-11-07|         IBM|113.85|\n",
      "|2002-02-06|         IBM|106.63|\n",
      "|2002-05-08|         IBM| 82.45|\n",
      "|2002-08-07|         IBM| 69.17|\n",
      "|2002-11-06|         IBM| 81.54|\n",
      "|2003-02-06|         IBM| 77.51|\n",
      "|2003-05-07|         IBM| 86.68|\n",
      "|2003-08-06|         IBM| 79.75|\n",
      "|2003-11-06|         IBM| 89.34|\n",
      "|2004-02-06|         IBM| 98.94|\n",
      "|2004-05-06|         IBM| 88.36|\n",
      "|2004-08-06|         IBM| 83.48|\n",
      "|2004-11-08|         IBM| 93.37|\n",
      "|2005-02-08|         IBM| 94.13|\n",
      "|2005-05-06|         IBM| 75.26|\n",
      "|2005-08-08|         IBM| 83.36|\n",
      "|2005-11-08|         IBM| 83.15|\n",
      "|2006-02-08|         IBM|  80.8|\n",
      "|2006-05-08|         IBM| 82.89|\n",
      "|2006-08-08|         IBM| 75.33|\n",
      "|2006-11-08|         IBM| 92.59|\n",
      "|2007-02-07|         IBM| 99.54|\n",
      "|2007-05-08|         IBM|103.29|\n",
      "|2007-08-08|         IBM|112.98|\n",
      "|2007-11-07|         IBM|111.08|\n",
      "|2008-02-06|         IBM|103.59|\n",
      "|2008-05-07|         IBM|124.14|\n",
      "|2008-08-06|         IBM|129.16|\n",
      "|2008-11-06|         IBM| 85.15|\n",
      "|2009-02-06|         IBM| 96.14|\n",
      "|2009-05-06|         IBM|104.62|\n",
      "|2009-08-06|         IBM|117.38|\n",
      "|2009-11-06|         IBM|123.49|\n",
      "|2010-02-08|         IBM|121.88|\n",
      "+----------+------------+------+\n",
      "\n",
      "------------the result of question2,duty 2--------------------\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 23:======================================>                   (2 + 1) / 3]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+---------------------+\n",
      "|year|stock_price_adj_close|\n",
      "+----+---------------------+\n",
      "|2006|    70.81063745019918|\n",
      "|2007|    128.2739043824701|\n",
      "|2008|   141.97901185770743|\n",
      "|2009|   146.81412698412706|\n",
      "|2010|             204.7216|\n",
      "+----+---------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import Row\n",
    "# sc = SparkContext(\"local\", \"duty 2\")\n",
    "stockrdd = sc.textFile('file:///home/allen/myspark/stock_small.csv')\n",
    "divrdd = sc.textFile('file:///home/allen/myspark/dividends_small.csv')\n",
    "stockrdd = stockrdd.map(lambda line:line.split(\",\"))\n",
    "divrdd = divrdd.map(lambda line:line.split(\",\"))\n",
    "sqlContext=SparkSession.builder.getOrCreate()\n",
    "stockrows = stockrdd.map(lambda p:Row(\n",
    "    exchange = p[0],\n",
    "    stock_symbol = p[1],\n",
    "    date =p[2],\n",
    "    open = p[3],\n",
    "    high = p[4],\n",
    "    low = p[5],\n",
    "    close = p[6],\n",
    "    volume = p[7],\n",
    "    close_adj = p[8]\n",
    "))\n",
    "stockdf = sqlContext.createDataFrame(stockrows)\n",
    "divrows = divrdd.map(lambda p:Row(\n",
    "    exchange = p[0],\n",
    "    symbol = p[1],\n",
    "    date =p[2],\n",
    "    dividend = p[3]\n",
    "))\n",
    "divdf = sqlContext.createDataFrame(divrows)\n",
    "# stockdf.printSchema()\n",
    "# stockdf.show()\n",
    "stockdf.registerTempTable(\"stocksm\")\n",
    "divdf.registerTempTable(\"div\")\n",
    "print('------------the result of question1,duty 2--------------------')\n",
    "sqlContext.sql(\"SELECT date,stock_symbol,close \\\n",
    "    FROM stocksm \\\n",
    "    WHERE stock_symbol = 'IBM' AND \\\n",
    "    date IN (SELECT date FROM div WHERE symbol = 'IBM')\" \n",
    "    ).show(100)\n",
    "\n",
    "print('------------the result of question2,duty 2--------------------')\n",
    "sqlContext.sql(\"SELECT year(date) year,avg(close_adj) stock_price_adj_close \\\n",
    "    FROM stocksm \\\n",
    "    WHERE stock_symbol = 'AAPL' \\\n",
    "    GROUP BY year\\\n",
    "    HAVING stock_price_adj_close>50 \\\n",
    "    ORDER BY year \"\n",
    "    ).show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## duty3\n",
    "\n",
    "使用的二分类方法是决策树，采用了pyspark.mllib.LabeledPoint \\ DecisionTree，数据集的划分方式为---训练集：验证集：测试集 = 7:1:2，模型检验方法是：计算测试集结果的均方误差，由于label值为0或1，均方误差小于0.5，说明有一定的预测性，但是不多。。。  \n",
    "\n",
    "代码的注释部分是对于决策树模型中模型参数的探索代码，可以测试哪种参数组合最为适宜，由于计算时间过长并且输出量超过了jubter的限制，所以进行了注释，但是截图会存储在gitee仓库中的result中。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 70:=============================>                            (1 + 1) / 2]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "训练集样本个数：992896验证集样本个数：141155测试集样本个数：283590\n",
      "均方误差RMSE=0.4205782185812733\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "from pyspark import SparkContext\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "from pyspark.mllib.tree import DecisionTree\n",
    "import numpy as np\n",
    "from pyspark.mllib.evaluation import RegressionMetrics\n",
    "import time\n",
    "# sc = SparkContext(\"local\", \"duty 3\")\n",
    "sc.setLogLevel(\"ERROR\")\n",
    "stockrdd = sc.textFile('file:///home/allen/myspark/stock_data.csv')\n",
    "header = stockrdd.first()\n",
    "rDate = stockrdd.filter(lambda x: x !=header)\n",
    "lines = rDate.map(lambda x: x.split(\",\"))\n",
    "def convert_float(v):\n",
    "    return float(v)\n",
    "def process_features(line):\n",
    "    \"\"\"处理特征，line为字段行\"\"\"\n",
    "    ## 处理余下的特征\n",
    "    Features = [convert_float(value) for value in line[3:7]]\n",
    "    # 返回拼接的总特征列表\n",
    "    return Features\n",
    "def process_label(line):\n",
    "    return float(line[-1])\n",
    "process_label(lines.first())\n",
    "labelpointRDD = lines.map(lambda r: LabeledPoint(process_label(r), process_features(r)))\n",
    "(trainData, validationData, testData) = labelpointRDD.randomSplit([7,1,2])\n",
    "trainData.persist()\n",
    "validationData.persist()\n",
    "testData.persist()\n",
    "model = DecisionTree.trainRegressor(trainData, categoricalFeaturesInfo={}, impurity=\"variance\", maxDepth=5, maxBins=32, \n",
    "                            minInstancesPerNode=1, minInfoGain=0.0)\n",
    "## 定义模型评估函数\n",
    "def RMSE(model, validationData):\n",
    "    ## 计算模型的准确率\n",
    "    predict = model.predict(validationData.map(lambda p:p.features))\n",
    "    ## 拼接预测值和实际值\n",
    "    predict_real = predict.zip(validationData.map(lambda p: p.label))\n",
    "    ## 计算均方误差\n",
    "    rmse = np.sqrt(predict_real.map(lambda p: (p[0]-p[1])**2).sum() / predict_real.count())\n",
    "    return rmse\n",
    "\n",
    "## 调用函数求模型在验证集上的准确率\n",
    "rmse =  RMSE(model, validationData)\n",
    "print(\"训练集样本个数：\"+str(trainData.count()) + \"验证集样本个数：\"+str(validationData.count())+ \"测试集样本个数：\"+str(testData.count()))\n",
    "print(\"均方误差RMSE=\"+str(rmse))\n",
    "\n",
    "\n",
    "# 创建trainEvaluateModel函数包含训练与评估功能，并计算训练评估的时间。\n",
    "\n",
    "# def trainEvaluateModel(trainData, validationData, maxDepthParm, maxBinsParm, minInstancesPerNodeParm, minInfoGainParm):\n",
    "#     startTime = time.time()\n",
    "#     ## 创建并训练模型\n",
    "#     model = DecisionTree.trainRegressor(trainData, categoricalFeaturesInfo={}, impurity=\"variance\", maxDepth=maxDepthParm, \n",
    "#                                         maxBins=maxBinsParm, minInstancesPerNode=minInstancesPerNodeParm, minInfoGain=minInfoGainParm)\n",
    "#     ## 计算RMSE\n",
    "#     rmse = RMSE(model, validationData)\n",
    "#     duration = time.time() - startTime   # 持续时间\n",
    "#     print(\"训练评估：参数\"+ \",  maxDepth=\"+str(maxDepthParm)+\",  maxBins=\"+str(maxBinsParm)+ \n",
    "#           \", minInstancesPerNode=\"+str(minInstancesPerNodeParm) +\", minInfoGainParm=\"+str(minInfoGainParm)+\"\\n\"\n",
    "#          \"===>消耗时间=\"+str(duration)+\",  均方误差RMSE=\"+str(rmse))\n",
    "#     return rmse, duration, maxDepthParm, maxBinsParm, minInstancesPerNodeParm, minInfoGainParm, model\n",
    "\n",
    "\n",
    "# ## 定义函数gridSearch网格搜索最佳参数组合\n",
    "# def gridSearch(trainData, validationData, maxDepthList, maxBinsList, minInstancesPerNodeList, minInfoGainList):\n",
    "#     metrics = [trainEvaluateModel(trainData, validationData, maxDepth, maxBins, minInstancesPerNode, minInfoGain)\n",
    "#           for maxDepth in maxDepthList\n",
    "#           for maxBins in maxBinsList\n",
    "#           for minInstancesPerNode in minInstancesPerNodeList\n",
    "#           for minInfoGain in minInfoGainList]\n",
    "#     # 按照RMSE从小到大排序，返回最小RMSE的参数组合\n",
    "#     sorted_metics = sorted(metrics, key=lambda k:k[0], reverse=False)\n",
    "#     best_parameters = sorted_metics[0]\n",
    "#     print(\"最佳参数组合：\"+\"maxDepth=\"+str( best_parameters[2]) + \n",
    "#          \",  maxBins=\"+str( best_parameters[3])+\",  minInstancesPerNode=\"+str( best_parameters[4])+\n",
    "#           \", minInfoGain=\"+str(best_parameters[5])+\"\\n\"+\n",
    "#          \",  均方误差RMSE=\"+str( best_parameters[0]))\n",
    "#     return  best_parameters\n",
    "# ## 参数组合\n",
    "# maxDepthList = [3, 5, 10,20,25]\n",
    "# maxBinsList = [30, 50,100,200]\n",
    "# minInstancesPerNodeList=[1,3,5,10,20]\n",
    "# minInfoGainList=[0.0,0.3,0.5]\n",
    "\n",
    "# ## 调用函数返回最佳参数组合\n",
    "# best_parameters = gridSearch(trainData, validationData, maxDepthList, maxBinsList, minInstancesPerNodeList, minInfoGainList)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.8.5 64-bit",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "e7370f93d1d0cde622a1f8e1c04877d8463912d04d973331ad4851f04de6915a"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
